package com.xigua.basicsAndApplication;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import java.lang.reflect.Type;
import java.util.Set;
import java.util.UUID;

/**
 * @author LiRongHua
 * @Title: ${file_name}
 * @Package ${package_name}
 * @Description: ${todo}
 * @date 2022/1/1321:20
 */
public class RedisDelayingQueue<T> {

    private static JedisPool jedisPool;
    static {
        JedisPoolConfig config = new JedisPoolConfig();
        config.setMaxTotal(10);
        config.setMaxWaitMillis(30000);
        jedisPool = new JedisPool(config,"120.76.116.216",6379,3000,"xg123456");

    }

    static class TaskItem<T> {
        public String id;
        public T msg;
    }

    private Jedis jedis;
    private String queueKey;

    //fastjson 序列化对象中存在generic类型时，需要使用TypeReference;
    private Type TaskType = new TypeReference<TaskItem>(){}.getType();

    public RedisDelayingQueue(Jedis jedis, String queueKey) {
        this.jedis = jedis;
        this.queueKey = queueKey;
    }

    public void delay(T msg){
        TaskItem<T> task = new TaskItem<T>();
        task.id  = UUID.randomUUID().toString();
        task.msg = msg;
        String s = JSON.toJSONString(task);
        jedis.zadd(queueKey,System.currentTimeMillis()+5000,s);

    }

    public void loop(){
        while (!Thread.interrupted()){
            Set<String> values = jedis.zrangeByScore(queueKey, 0, System.currentTimeMillis(), 0, 1);
            if (values.isEmpty()){
                try {
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        break;
                    }
                    continue;
                }finally {

                }
            }
            String s = values.iterator().next();
            if (jedis.zrem(queueKey,s) > 0){
                TaskItem task = JSON.parseObject(s, TaskType);
                this.handleMsg((T) task.msg);
            }
        }
    }

    private void handleMsg(T msg) {
        System.out.println(msg);
    }

    public static void main(String[] args) {
        Jedis jedis = jedisPool.getResource();
        RedisDelayingQueue<Object> queue = new RedisDelayingQueue(jedis,"q-demo");
        Thread prodcuer =   new Thread(()->{
            for (int i = 0; i < 10; i++) {
                queue.delay("codehole"+i);
            }
        });

        Thread comsumer = new Thread(()->{
           queue .loop();
        });

        prodcuer.start();
        comsumer.start();

        try {
            prodcuer.join();
            Thread.sleep(6000);
            comsumer.interrupt();
            comsumer.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }


}
